package com.swak.reactivex.transport;

import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.concurrent.atomic.AtomicLong;

import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.experimental.Accessors;

/**
 * 
 * 通过此消息对象来控制两端通信；
 * 
 * @author DELL
 */
@Getter
@Setter
@Accessors(chain = true)
@ToString
public class Message implements Serializable {

	private static final long serialVersionUID = 1L;

	/**
	 * 请求
	 */
	public static final byte Request = 1 << 0;

	/**
	 * 响应
	 */
	public static final byte Response = 1 << 1;

	/**
	 * 需要响应
	 */
	public static final byte TwoWay = 1 << 2;

	/**
	 * 心跳事件
	 */
	public static final byte Heartbeat = 1 << 3;

	/**
	 * 使用Gzip压缩
	 */
	public static final byte Gzip = 1 << 4;

	/**
	 * 事件：编解码时不考虑ID位
	 */
	public static final byte Event = 1 << 5;

	/**
	 * 数据流： 没有后面的 序列化 - 状态 - ID等
	 */
	public static final byte Stream = 1 << 6;

	/**
	 * 状态位 --  业务异常也是 OK 状态
	 */
	public static final byte OK = 1;

	/**
	 * 状态位 -- 传输过程、编解码过程中的异常、超时异常（和框架有关）
	 */
	public static final byte CLIENT_TIMEOUT = 2;
	public static final byte SERVER_TIMEOUT = 3;
	public static final byte CHANNEL_INACTIVE = 4;
	public static final byte BAD_REQUEST = 5;
	public static final byte BAD_RESPONSE = 6;
	public static final byte SERVICE_NOT_FOUND = 7;
	public static final byte SERVICE_ERROR = 8;
	public static final byte SERVER_ERROR = 9;
	public static final byte CLIENT_ERROR = 10;

	/**
	 * 基本属性
	 */
	public static final String Id = "id";
	public static final String Status = "status";
	public static final String Option = "option";
	public static final String Data = "data";
	private static final AtomicLong INVOKE_ID = new AtomicLong(1);

	private long id;
	private byte status = OK;
	private byte option = 0; // 功能位：0~5 位； 序列化：6~7
	private int headerLen; // 头部长度
	private int bodyLen; // 数据长度
	private Object data;
	private Throwable error;

	/**
	 * UDP 发送数据时，需要指定发送数据的地址
	 */
	private transient InetSocketAddress address;

	public Message() {
	}

	public Message(long id) {
		this.id = id;
	}

	/**
	 * 是否请求
	 * 
	 * @return
	 */
	public boolean isRequest() {
		return (this.option & Request) == Request;
	}

	/**
	 * 是否请求
	 * 
	 * @return
	 */
	public boolean isResponse() {
		return (this.option & Response) == Response;
	}

	/**
	 * 是否需要响应
	 * 
	 * @return
	 */
	public boolean isTwoWay() {
		return (this.option & TwoWay) == TwoWay;
	}

	/**
	 * 是否心跳事件
	 * 
	 * @return
	 */
	public boolean isHeartbeat() {
		return (this.option & Heartbeat) == Heartbeat;
	}

	/**
	 * 是否使用Gzip压缩
	 * 
	 * @return
	 */
	public boolean isUseGzip() {
		return (this.option & Gzip) == Gzip;
	}

	/**
	 * 是否 事件
	 * 
	 * @return
	 */
	public boolean isEvent() {
		return (this.option & Event) == Event;
	}

	/**
	 * 是否 流数据： 不需要序列化，不需要状态 ，不需要ID； data 为 ByteBuf
	 * 
	 * @return
	 */
	public boolean isStream() {
		return (this.option & Stream) == Stream;
	}

	/**
	 * 添加选项
	 * 
	 * @param option
	 * @return
	 */
	public Message addOption(byte option) {
		this.option = (byte) (this.option | option);
		return this;
	}

	/**
	 * 删除选项
	 * 
	 * @param option
	 * @return
	 */
	public Message delOption(byte option) {
		this.option = (byte) (this.option & (~option));
		return this;
	}

	public Message address(String host, int port) {
		this.address = new InetSocketAddress(host, port);
		return this;
	}

	/**
	 * 发送消息
	 * 
	 * @return
	 */
	public static Message forMessage() {
		return new Message();
	}

	/**
	 * 发送消息
	 * 
	 * @return
	 */
	public static Message forMessage(Object msg) {
		if (msg instanceof Message) {
			Message message = (Message) msg;
			return new Message().setData(message.getData()).setError(message.getError()).setId(message.getId())
					.setOption(message.getOption()).setStatus(message.getStatus());
		}
		return forMessage().setData(msg);
	}

	/**
	 * 事件： 不需要响应，也没有ID字段
	 * 
	 * @return
	 */
	public static Message forEvent() {
		Message request = new Message();
		request.option = (byte) (request.option | Event);
		return request;
	}

	/**
	 * 流处理： 不需要响应，也没有ID字段
	 * 
	 * @return
	 */
	public static Message forStream() {
		Message request = new Message();
		request.option = (byte) (request.option | Stream);
		return request;
	}

	/**
	 * 请求: 默认需要响应
	 * 
	 * @return
	 */
	public static Message forRequest() {
		Message request = new Message(INVOKE_ID.getAndIncrement());
		request.option = (byte) (request.option | Request | TwoWay);
		return request;
	}

	/**
	 * 发送消息
	 * 
	 * @return
	 */
	public static Message forRequest(Object msg) {
		if (msg instanceof Message) {
			Message message = (Message) msg;
			return new Message().setData(message.getData()).setError(message.getError()).setId(message.getId())
					.setOption(message.getOption()).setStatus(message.getStatus());
		}
		return forRequest().setData(msg);
	}

	/**
	 * 请求: 默认需要响应
	 * 
	 * @return
	 */
	public static Message forResponse(Message request) {
		Message response = new Message(request.getId());
		response.option = (byte) (response.option | Response);
		return response;
	}

	/**
	 * 消息是否请求
	 * 
	 * @param message
	 * @return
	 */
	public static boolean isResquest(Object message) {
		return message instanceof Message && ((Message) message).isRequest();
	}

	/**
	 * 消息是否响应
	 * 
	 * @param message
	 * @return
	 */
	public static boolean isResponse(Object message) {
		return message instanceof Message && ((Message) message).isResponse();
	}

	/**
	 * 消息事件
	 * 
	 * @param message
	 * @return
	 */
	public static boolean isEvent(Object message) {
		return message instanceof Message && ((Message) message).isEvent();
	}

	/**
	 * 流数据：比较大的数据
	 * 
	 * @param message
	 * @return
	 */
	public static boolean isStream(Object message) {
		return message instanceof Message && ((Message) message).isStream();
	}
}
